package cn.com.sparkle.paxos.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.ProtobufProtocolBuilder;
import cn.com.sparkle.paxos.client.ConnectConfig.ConnectEvent;
import cn.com.sparkle.paxos.client.net.GetNewsSenatorsCallBack;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.GetNewsSenatorsRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.GetNewsSenatorsResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.MajorSender;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.raptor.core.handler.IoHandler;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketClient;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketConfigure;

public class CommandProcessor implements Runnable {
	private final static Logger logger = Logger
			.getLogger(CommandProcessor.class);
	// private final static GetNewsSenatorsRequest GET_NEWS_SENATORS_REQUEST =
	// new GetNewsSenatorsRequest();

	private static NioSocketClient nioSocketClient;
	private static ClientHandler handler;
	private static IoHandler ioHandler;
	private static ReConnectThread reConnectThread = new ReConnectThread();

	private volatile String[] senator = {};
	private volatile NetNode node;
	private volatile ConnectConfig firstPriorConfig;
	private volatile HashMap<String, NetNode> activeNode = new HashMap<String, NetNode>();

	private ArrayBlockingQueue<Command> commandQueue;
	private LinkedBlockingQueue<Command> runningCommand;
	// private LinkedBlockingQueue<Command> ru

	private final int maxRunningSize;
	// private final int maxWaitRunSize;

	private ReentrantLock lock = new ReentrantLock();
	private Condition wakeupCondition = lock.newCondition();

	private ReentrantLock nodeChangeLock = new ReentrantLock();

	private volatile boolean isWakeup = false;

	private boolean isPromiseMaster;

	private static class CommandCallback implements CallBack {
		private Command c;
		private CommandProcessor processor;

		public CommandCallback(Command c, CommandProcessor processor) {
			this.c = c;
			this.processor = processor;
		}

		@Override
		public void call(NetNode nnode, MessagePackage value) {
			c.finish(nnode, value);
			Command comm;
			while ((comm = processor.runningCommand.peek()) != null) {
				if (comm.isFinish()) {
					processor.runningCommand.poll();
					processor.wakeup();
				} else {
					break;
				}
			}
		}

		@Override
		public void fail(NetNode nnode) {
			// TODO Auto-generated method stub

		}

	}

	// private CallBack callback = new CallBack() {
	//
	// @Override
	// public void call(NetNode nnode, MessagePackage value) {
	//
	// finish(nnode,value);
	// }
	//
	// @Override
	// public void fail(NetNode nnode) {
	// }
	// };

	private ConnectEvent connectEvent = new ConnectEvent() {
		@Override
		public void disconnect(String address, NetNode node) {
			synchronized (this) {
				HashMap<String, NetNode> newSet = new HashMap<String, NetNode>();
				newSet.putAll(activeNode);
				newSet.remove(address);
				activeNode = newSet;
				node.close();
			}
		}

		@Override
		public void connect(String address, NetNode node) {
			synchronized (this) {
				HashMap<String, NetNode> newSet = new HashMap<String, NetNode>();
				newSet.putAll(activeNode);
				newSet.put(address, node);
				activeNode = newSet;
			}
		}
	};

	public void runDeamon() {
		Thread t = new Thread(this);
		t.setName("CommandProcessor");
		t.start();
	}

	public static void init(int threadCount) throws IOException {
		NioSocketConfigure nsc = new NioSocketConfigure();
		nsc.setProcessorNum(1);
		nsc.setTcpNoDelay(true);
		nsc.setCycleRecieveBuffCellSize(2000);
		reConnectThread.startThread();
		nioSocketClient = new NioSocketClient(nsc, "userclient");
		handler = new ClientHandler(nioSocketClient, reConnectThread);
		ioHandler = new MultiThreadProtecolHandler(1000, 16 * 1024,
				threadCount, threadCount, 60, TimeUnit.SECONDS,
				ProtobufProtocolBuilder.build(), handler, "userclient");
		handler.setHandler(ioHandler);
	}

	public CommandProcessor(String[] senator, int maxRunningSize,
			int maxWaitRunSize, boolean isPromiseMaster) throws Exception {
		this.isPromiseMaster = isPromiseMaster;
		this.senator = senator;
		this.maxRunningSize = maxRunningSize;
		runningCommand = new LinkedBlockingQueue<Command>(maxRunningSize);
		commandQueue = new ArrayBlockingQueue<Command>(maxWaitRunSize);

		// for (String aa : senator) {
		String[] tmp = senator[0].split(":");
		InetSocketAddress iAddress = new InetSocketAddress(tmp[0],
				Integer.parseInt(tmp[1]));
		firstPriorConfig = new ConnectConfig(senator[0], true, isPromiseMaster,
				connectEvent);
		boolean isSuccess = nioSocketClient.connect(iAddress, ioHandler,
				firstPriorConfig).get();
		if (isSuccess) {
			firstPriorConfig.waitOnOpenEvent();
		}
		// }
	}

	public void changeSenator(String[] senator) throws Exception {
		try {
			nodeChangeLock.lock();
			if (!senator[0].equals(this.senator[0])) {
				firstPriorConfig.setAutoReConnect(false);
				firstPriorConfig = new ConnectConfig(senator[0], true,
						isPromiseMaster, connectEvent);
				String[] tmp = senator[0].split(":");
				InetSocketAddress iAddress = new InetSocketAddress(tmp[0],
						Integer.parseInt(tmp[1]));
				boolean isSuccess = nioSocketClient.connect(iAddress,
						ioHandler, firstPriorConfig).get();
				if (isSuccess) {
					firstPriorConfig.waitOnOpenEvent();
				}
			}

			NetNode n = node;
			if (n != null) {
				n.close();
			}
			this.senator = senator;
		} finally {
			nodeChangeLock.unlock();
		}
	}

	public void wakeup() {
		try {
			lock.lock();
			isWakeup = true;
			wakeupCondition.signal();
		} finally {
			lock.unlock();
		}
	}

	int testCount = 0;
	long ct;

	public void run() {
		try {
			while (true) {
				try {
					lock.lock();
					if (!isWakeup) {
						wakeupCondition.await();
					}
					isWakeup = false;
				} finally {
					lock.unlock();
				}
				while (node == null) {
					// reconnect master
					reConnect();
					// redo
					for (Command cc : runningCommand) {
						if (!cc.isFinish()) {
							if (!node.write(cc.getRequest(),
									new CommandCallback(cc, this))) {
								logger.error("master disconnected!");
								node = null;
								Thread.sleep(1000);
								break;
							}
						}
					}
				}

				while (commandQueue.peek() != null
						&& runningCommand.size() != maxRunningSize) {
					if (testCount == 0) {
						ct = System.currentTimeMillis();
					}
					if (++testCount % 1000 == 0) {
						logger.debug("send " + testCount);
					}
					Command c = commandQueue.poll();
					if (c.hasCallback()) {
						runningCommand.add(c);
					}
					if (!node.write(c.getRequest(),
							new CommandCallback(c, this))) {
						logger.error("master disconnect!");
						node = null;
						wakeup();
						break;
					}

				}
				// logger.debug("runningCommand size:" + runningCommand.size());
			}

		} catch (InterruptedException e) {
			throw new RuntimeException("not support interruptedException", e);
		}
	};

	// public void finish(NetNode node,MessagePackage response) {
	// Command c = null;
	// c = runningCommand.poll();
	// wakeup();
	// c.getCallBack().response(node,response);
	// }

	public boolean addCommand(Command c, int timeout, TimeUnit timeUnit)
			throws InterruptedException {
		if (timeout == 0) {
			commandQueue.put(c);
			wakeup();
			return true;
		} else if (commandQueue.offer(c, timeout, timeUnit)) {
			// logger.debug("add command");
			wakeup();
			return true;
		} else {
			return false;
		}
	}

	public NetNode getNode() {
		return node;
	}

	private void reConnect() throws InterruptedException {
		try {
			NetNode node = null;
			while (node == null) {
				String[] senator = this.senator;
				HashMap<String, NetNode> activeSenators = activeNode;
				node = activeSenators.get(senator[0]);
				if (node != null) {
					logger.info("connect to " + senator[0]);
					// close other senator
					for (Entry<String, NetNode> e : activeSenators.entrySet()) {
						if (!e.getKey().equals(senator[0])) {
							e.getValue().close();
						}
					}
				} else {
					if (senator.length > 1) {
						for (int i = 1; i < senator.length; ++i) {
							node = activeNode.get(senator[i]);
							if (node == null) {
								String[] tmp = senator[i].split(":");
								InetSocketAddress iAddress = new InetSocketAddress(
										tmp[0], Integer.parseInt(tmp[1]));

								ConnectConfig connectConfig = new ConnectConfig(
										senator[i], false, isPromiseMaster,
										connectEvent);
								boolean isSuccess = nioSocketClient.connect(
										iAddress, ioHandler, connectConfig)
										.get();
								if (isSuccess) {
									connectConfig.waitOnOpenEvent();
									logger.info("connect to " + senator[i]);
								}
							}
							node = activeNode.get(senator[i]);
							if (node != null) {
								break;
							}
						}
					}
				}

				if (node == null) {
					Thread.sleep(2000);
					logger.error("find master position failed!");
				} else {
					try {
						nodeChangeLock.lock();
						if (senator == this.senator) {
							this.node = node;
						} else {
							node.close();
						}
					} finally {
						nodeChangeLock.unlock();
					}
				}
			}
		} catch (ExecutionException e) {
			logger.error("fatal", e);
		} catch (Exception e) {
			logger.error("fatal", e);
		}
	}
}
